#include <iostream>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <chrono>
#include <atomic>
#include <vector>
#include <sstream>
#include <iomanip>
#include <csignal>

template<typename T>
class AdaptiveBoundedQueue {
public:
    // 构造函数,支持配置初始水位和超时时间
    AdaptiveBoundedQueue(
        size_t initialHighWaterMark = 1000,
        size_t initialLowWaterMark = 100,
        double adjustFactor = 1.5,
        size_t maxWaterMark = 10000,
        std::chrono::milliseconds timeout = std::chrono::seconds(5))
        : highWaterMark_(initialHighWaterMark)
        , lowWaterMark_(initialLowWaterMark)
        , maxWaterMark_(maxWaterMark)
        , adjustFactor_(adjustFactor)
        , timeout_(timeout)
        , pause_(false)
        , pressureCount_(0)
        , totalItemsProcessed_(0)
        , lastAdjustTime_(std::chrono::steady_clock::now()) {
    }

    // 放入数据,支持阻塞和非阻塞模式
    bool put(const T& item, bool blocking = true) {
        std::unique_lock<std::mutex> lock(mutex_);
        
        if (blocking) {
            while (pause_) {
                auto status = producerCondition_.wait_for(lock, timeout_);
                if (status == std::cv_status::timeout) {
                    std::cout << "Producer timeout after " 
                             << timeout_.count() << "ms" << std::endl;
                    return false;
                }
            }
        } else if (pause_) {
            return false;
        }

        queue_.push(item);
        totalItemsProcessed_++;
        
        updatePressure(lock);
        consumerCondition_.notify_one();
        
        return true;
    }

    // 取出数据,支持阻塞和非阻塞模式
    bool take(T& item, bool blocking = true) {
        std::unique_lock<std::mutex> lock(mutex_);
        
        if (blocking) {
            while (queue_.empty()) {
                auto status = consumerCondition_.wait_for(lock, timeout_);
                if (status == std::cv_status::timeout) {
                    std::cout << "Consumer timeout after " 
                             << timeout_.count() << "ms" << std::endl;
                    return false;
                }
            }
        } else if (queue_.empty()) {
            return false;
        }

        item = queue_.front();
        queue_.pop();

        checkPressureRelease(lock);
        
        return true;
    }

    // 获取队列统计信息
    struct QueueStats {
        size_t currentSize;
        size_t highWaterMark;
        size_t lowWaterMark;
        bool isPaused;
        size_t pressureCount;
        size_t totalProcessed;
        
        std::string toString() const {
            std::stringstream ss;
            ss << "Size: " << std::setw(6) << currentSize
               << " H/L: " << std::setw(6) << highWaterMark 
               << "/" << std::setw(6) << lowWaterMark
               << " Paused: " << (isPaused ? "Yes" : "No")
               << " Pressure: " << std::setw(3) << pressureCount
               << " Total: " << std::setw(8) << totalProcessed;
            return ss.str();
        }
    };

    QueueStats getStats() const {
        std::lock_guard<std::mutex> lock(mutex_);
        return {
            queue_.size(),
            highWaterMark_,
            lowWaterMark_,
            pause_,
            pressureCount_,
            totalItemsProcessed_
        };
    }

private:
    void updatePressure(std::unique_lock<std::mutex>& lock) {
        if (queue_.size() >= highWaterMark_) {
            pressureCount_++;
            
            auto now = std::chrono::steady_clock::now();
            if (pressureCount_ > 3 && 
                std::chrono::duration_cast<std::chrono::seconds>
                (now - lastAdjustTime_).count() > 5) {
                
                size_t newHighWaterMark = std::min(
                    static_cast<size_t>(highWaterMark_ * adjustFactor_),
                    maxWaterMark_);
                
                if (newHighWaterMark > highWaterMark_) {
                    highWaterMark_ = newHighWaterMark;
                    lowWaterMark_ = highWaterMark_ / 10;
                    lastAdjustTime_ = now;
                    
                    std::cout << "Increasing water marks to: " 
                             << lowWaterMark_ << "/" << highWaterMark_ 
                             << std::endl;
                }
            }

            if (!pause_) {
                std::cout << "High water mark reached! Queue size: " 
                         << queue_.size() << " Current marks: "
                         << lowWaterMark_ << "/" << highWaterMark_ 
                         << std::endl;
                pause_ = true;
            }
        }
    }

    void checkPressureRelease(std::unique_lock<std::mutex>& lock) {
        if (queue_.size() <= lowWaterMark_) {
            if (pause_) {
                pause_ = false;
                std::cout << "Low water mark reached! Queue size: " 
                         << queue_.size() << " Current marks: "
                         << lowWaterMark_ << "/" << highWaterMark_ 
                         << std::endl;
                producerCondition_.notify_all();
            }

            auto now = std::chrono::steady_clock::now();
            if (pressureCount_ == 0 && 
                std::chrono::duration_cast<std::chrono::seconds>
                (now - lastAdjustTime_).count() > 10 &&
                highWaterMark_ > 1000) {
                
                highWaterMark_ = std::max(
                    static_cast<size_t>(highWaterMark_ / adjustFactor_),
                    static_cast<size_t>(1000));
                lowWaterMark_ = highWaterMark_ / 10;
                lastAdjustTime_ = now;
                
                std::cout << "Decreasing water marks to: " 
                         << lowWaterMark_ << "/" << highWaterMark_ 
                         << std::endl;
            }
        }
    }

    std::queue<T> queue_;
    mutable std::mutex mutex_;
    std::condition_variable producerCondition_;
    std::condition_variable consumerCondition_;
    
    size_t highWaterMark_;
    size_t lowWaterMark_;
    const size_t maxWaterMark_;
    const double adjustFactor_;
    const std::chrono::milliseconds timeout_;
    
    bool pause_;
    size_t pressureCount_;
    size_t totalItemsProcessed_;
    std::chrono::steady_clock::time_point lastAdjustTime_;
};

// 全局变量用于信号处理
std::atomic<bool> g_running(true);

void signalHandler(int signum) {
    std::cout << "Interrupt signal (" << signum << ") received.\n";
    g_running = false;
}

void producer(AdaptiveBoundedQueue<int>& queue, int id,
             std::atomic<int>& produceRate) {
    int count = 0;
    while (g_running) {
        if (!queue.put(count++)) {
            std::cout << "Producer " << id << " failed to put item" 
                     << std::endl;
        }
        std::this_thread::sleep_for(
            std::chrono::milliseconds(produceRate.load()));
    }
}

void consumer(AdaptiveBoundedQueue<int>& queue, int id,
             std::atomic<int>& consumeRate) {
    int item;
    while (g_running) {
        if (queue.take(item)) {
            std::this_thread::sleep_for(
                std::chrono::milliseconds(consumeRate.load()));
        }
    }
}

void monitor(AdaptiveBoundedQueue<int>& queue) {
    while (g_running) {
        auto stats = queue.getStats();
        std::cout << stats.toString() << std::endl;
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
}

int main() {
    signal(SIGINT, signalHandler);
    
    AdaptiveBoundedQueue<int> queue(1000, 100);
    std::atomic<int> produceRate(10);
    std::atomic<int> consumeRate(50);

    std::vector<std::thread> producers;
    std::vector<std::thread> consumers;
    
    // 启动监控线程
    std::thread monitorThread(monitor, std::ref(queue));

    // 启动生产者和消费者
    for (int i = 0; i < 3; ++i) {
        producers.emplace_back(producer, std::ref(queue), i, 
                             std::ref(produceRate));
    }
    
    for (int i = 0; i < 1; ++i) {
        consumers.emplace_back(consumer, std::ref(queue), i,
                             std::ref(consumeRate));
    }

    // 模拟负载变化
    for (int phase = 0; phase < 3 && g_running; ++phase) {
        std::this_thread::sleep_for(std::chrono::seconds(10));
        
        switch (phase) {
            case 0:
                std::cout << "\nPhase 1: Increasing produce rate\n" 
                         << std::endl;
                produceRate.store(5);
                break;
            case 1:
                std::cout << "\nPhase 2: Decreasing consume rate\n" 
                         << std::endl;
                consumeRate.store(100);
                break;
            case 2:
                std::cout << "\nPhase 3: Returning to normal\n" 
                         << std::endl;
                produceRate.store(10);
                consumeRate.store(50);
                break;
        }
    }

    // 清理资源
    g_running = false;
    monitorThread.join();
    
    for (auto& t : producers) {
        t.join();
    }
    for (auto& t : consumers) {
        t.join();
    }

    return 0;
}